package guda.mrpc.core.transport.callback;

import guda.mrpc.core.RequestKey;
import guda.mrpc.core.client.ConnectionFactory;
import guda.mrpc.core.registry.ServiceMeta;
import guda.mrpc.core.serialize.ObjectSerialize;
import guda.mrpc.core.serialize.impl.DefaultSerialize;
import guda.mrpc.core.transport.ClientHandler;
import guda.mrpc.core.transport.tcp.TcpClientChannelInitializer;
import guda.mrpc.core.transport.tcp.TcpClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by well on 17/5/23.
 */
public class RequestCallback implements Callable<Boolean> {

    private Logger log = LoggerFactory.getLogger(RequestCallback.class);

    protected EventLoopGroup eventLoopGroup = null;
    protected ServiceMeta serviceMeta = null;
    protected ObjectSerialize objectSerialize;

    private Lock lock = new ReentrantLock();

    private static final byte DEFAULT_TRCRY = 10;

    // 10次重连机会,超过后关闭客户端连接,成功后清零
    private AtomicInteger retries = new AtomicInteger(1);

    public RequestCallback(ServiceMeta serviceMeta, EventLoopGroup eventLoopGroup) {
        this.serviceMeta = serviceMeta;
        this.eventLoopGroup = eventLoopGroup;
        this.objectSerialize = new DefaultSerialize();
    }

    @Override
    public Boolean call() throws Exception {
        connectServer(new Bootstrap(), eventLoopGroup);
        return Boolean.TRUE;
    }

    private void connectServer(Bootstrap b, EventLoopGroup eventLoopGroup) {
        b.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true);

        b.handler(new TcpClientChannelInitializer(objectSerialize));
        SocketAddress socketAddress = new InetSocketAddress(serviceMeta.getHost(),serviceMeta.getPort());
        ChannelFuture channelFuture = b.connect(socketAddress);
        channelFuture.addListener(new ConnectionListener());
    }


    class ConnectionListener implements ChannelFutureListener {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            lock.lock();
            if (future.isSuccess()) {
                retries.set(0);
                if(log.isDebugEnabled()){
                    log.debug("Client connect success");
                }
                if (null != serviceMeta ) {
                    Class<? extends ClientHandler> clientHandler = TcpClientHandler.class;
                    ClientHandler handler = future.channel().pipeline().get(clientHandler);
                    handler.setServiceMeta(serviceMeta);
                    ConnectionFactory.instance().addClientHandler(serviceMeta.getRequestKey(), handler);

                }
            } else {
                log.warn("Client reconnect [{}] ({})", serviceMeta.getAddress(), retries.get());
                final EventLoop loop = future.channel().eventLoop();
                loop.schedule(() -> {
                    if (retries.get() >= DEFAULT_TRCRY) {
                        future.channel().close();
                        ConnectionFactory.instance().remove(serviceMeta);
                        log.warn("Client channel connect fail, closed.");
                    } else {
                        connectServer(new Bootstrap(), loop);
                        retries.incrementAndGet();
                    }
                }, 5L*(2<<(retries.get()+1)), TimeUnit.SECONDS);
            }
            lock.unlock();
        }
    }
}
